package com.afdor.rws.event.support;

import com.afdor.rws.annotation.extension.Extension;
import com.afdor.rws.core.extension.ExtensionLoader;
import com.afdor.rws.core.strategy.IdGeneratorStrategy;
import com.afdor.rws.event.AggregationEvent;
import com.afdor.rws.event.Event;
import com.afdor.rws.event.EventBus;
import com.afdor.rws.event.EventPersistenceManager;
import com.afdor.rws.event.EventSource;
import com.afdor.rws.event.EventSourceManager;
import com.afdor.rws.event.channel.client.consumer.handler.EventHandler;
import com.afdor.rws.event.config.EventConfigUtils;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.NonNull;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * 事件中线默认实现
 *
 * @author 悭梵
 * @date Created in 2018-08-14 15:08
 */
@Slf4j
@ToString
@EqualsAndHashCode
@Extension(value = "default", order = Extension.NORMAL_ORDER - 100, interfaceClass = EventBus.class)
public class DefaultEventBus implements EventBus {

    private volatile boolean init = Boolean.FALSE;
    /**
     * 事件源管理器
     */
    @Getter
    private volatile EventSourceManager eventSourceManager;

    /**
     * 事件持久化管理器
     */
    @Getter
    private volatile EventPersistenceManager eventPersistenceManager;
    /**
     * 事物提交队列管理器
     */
    private ConcurrentMap<String, TransactionItem> eventTransactionMap = new ConcurrentHashMap<String, TransactionItem>(128);

    public DefaultEventBus() {
        /* jvm 退出执行方法 */
        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                if (eventSourceManager != null) {
                    eventSourceManager.destroy();
                    eventSourceManager = null;
                }
                if (eventPersistenceManager != null) {
                    eventPersistenceManager.destroy();
                    eventPersistenceManager = null;
                }
            }
        });
    }

    @Override
    public synchronized EventBus init() {
        if (!init) {
            // 注册默认的广播事件处理器和骨架事件处理器
            ExtensionLoader<EventHandler> eventHandlerExtensionLoader = ExtensionLoader.getInstance(EventHandler.class);
            String broadcastEventHandler = EventConfigUtils.getProperty(EventConfigUtils.EVENT_CONSUMER_HANDLER_DEFAULT_BROADCAST_NAME);
            String aggregationRootEventHandler = EventConfigUtils.getProperty(EventConfigUtils.EVENT_CONSUMER_HANDLER_DEFAULT_AGGREGATION_ROOT_NAME);
            // 空事件处理器
            EventHandler emptyEventHandler = event -> log.debug("<== " + event.toString());
            // 注册默认的广播事件处理器
            if (StringUtils.isBlank(broadcastEventHandler)) {
                registerHandler(Event.Type.BROADCAST.getDefaultSourceName(), emptyEventHandler);
            } else {
                registerHandler(Event.Type.BROADCAST.getDefaultSourceName(), eventHandlerExtensionLoader.getExtension(broadcastEventHandler));
            }
            // 注册默认的骨架事件处理器
            if (StringUtils.isBlank(aggregationRootEventHandler)) {
                registerHandler(Event.Type.AGGREGATION_ROOT.getDefaultSourceName(), emptyEventHandler);
            } else {
                registerHandler(Event.Type.AGGREGATION_ROOT.getDefaultSourceName(), eventHandlerExtensionLoader.getExtension(aggregationRootEventHandler));
            }
            init = Boolean.TRUE;
        }
        return this;
    }

    @Override
    public DefaultEventBus setEventSourceManager(@NonNull EventSourceManager eventSourceManager) {
        this.eventSourceManager = eventSourceManager;
        return this;
    }

    @Override
    public DefaultEventBus setEventPersistenceManager(EventPersistenceManager eventPersistenceManager) {
        this.eventPersistenceManager = eventPersistenceManager;
        return this;
    }

    @Override
    public EventBus registerHandler(@NonNull String eventSourceName, @NonNull EventHandler eventHandler) {
        checkup();
        EventSource eventSource = (EventSource) Optional.ofNullable(getEventSourceManager().getEventSource(eventSourceName))
                .orElseThrow(() -> new NullPointerException("未获取到事件处理器" + eventSourceName + "配置项"));

        Optional.ofNullable(eventSource.getChannelProvider())
                .orElseThrow(() -> new NullPointerException("事件处理器对应的发送通道为空"))
                .subscribe(eventSourceName, eventHandler);
        return this;
    }

    @Override
    public EventBus sendAndCommitEvent(@NonNull Event event) throws Exception {
        String transactionId = newTransaction();
        addEvent(transactionId, event);
        commit(transactionId);
        return this;
    }

    @NonNull
    @Override
    public String newTransaction() {
        checkup();

        ExtensionLoader<IdGeneratorStrategy> idGeneratorStrategyExtensionLoader = ExtensionLoader.getInstance(IdGeneratorStrategy.class);
        // 获取配置的Id生成策略
        String transactionId = String.valueOf(idGeneratorStrategyExtensionLoader.getExtensionOrDefault(EventConfigUtils.getProperty(EventConfigUtils.EVENT_EVENT_BUS_TRANSACTION_ID)).generate());
        if (eventTransactionMap.containsKey(transactionId)) {
            return newTransaction();
        } else {
            eventTransactionMap.putIfAbsent(transactionId, new TransactionItem());
        }
        return transactionId;
    }

    @Override
    public EventBus addEvent(@NonNull String transactionId, @NonNull Event event) throws IllegalAccessException {
        TransactionItem transactionItem = getTransactionItem(transactionId);
        if (transactionItem == null) {
            throw new IllegalStateException("the specified transaction is not turned on or has been executed. please open a new transaction first . ");
        }
        transactionItem.getAggregationEvent().addEvent(event);
        return this;
    }

    @Override
    public void commit(@NonNull String transactionId) throws Exception {
        checkup();
        TransactionItem transactionItem = eventTransactionMap.remove(transactionId);
        if (transactionItem != null) {
            transactionItem.commit();
        } else {
            throw new IllegalStateException("the specified transaction is not turned on or has been executed.");
        }
    }

    @Override
    public void rollback(@NonNull String transactionId) throws Exception {
        TransactionItem transactionItem = eventTransactionMap.remove(transactionId);
        if (transactionItem != null) {
            transactionItem.rollback();
        } else {
            throw new IllegalStateException("the specified transaction is not turned on or has been executed.");
        }
    }

    private void checkup() {
        if (eventSourceManager == null) {
            new NullPointerException("事件源管理器不能为空");
        }
        if (eventPersistenceManager == null) {
            // 默认事件持久化管理器
            eventPersistenceManager = ExtensionLoader
                    .getInstance(EventPersistenceManager.class)
                    .getExtensionOrDefault(EventConfigUtils.getProperty(EventConfigUtils.EVENT_EVENT_BUS_NAME));
        }
    }

    /**
     * 回去事件执行通道，发送事件信息
     *
     * @param event
     */
    private void sendEvent(@NonNull Event event) {
        if (event instanceof AggregationEvent) {
            AggregationEvent aggregationEvent = (AggregationEvent) event;

            String isSendAggregationEvent = EventConfigUtils.getProperty(EventConfigUtils.EVENT_EVENT_BUS_IS_SEND_AGGREGATION_ROOT_EVENT, "false");
            if (Boolean.valueOf(isSendAggregationEvent)) {
                /* 聚合事件优先发送 */
                EventSource aggregationEventSource = (EventSource) Optional.ofNullable(getEventSourceManager().getEventSource(event.getEventSourceName()))
                        .orElseThrow(() -> new NullPointerException("未获取到事件处理器配置项"));

                Optional.ofNullable(aggregationEventSource.getChannelProvider())
                        .orElseThrow(() -> new NullPointerException("事件处理器对应的发送通道为空"))
                        .sendEvent(event);
            }

            /* 循环迭代，发送子事件 */
            Optional.ofNullable(aggregationEvent.getSource())
                    .orElseThrow(() -> new NullPointerException("事件队列为空"))
                    .stream().forEach(item -> {
                EventSource eventSource = (EventSource) Optional.ofNullable(getEventSourceManager().getEventSource(item.getEventSourceName()))
                        .orElseThrow(() -> new NullPointerException("未获取到事件处理器配置项"));

                Optional.ofNullable(eventSource.getChannelProvider())
                        .orElseThrow(() -> new NullPointerException("事件处理器对应的发送通道为空"))
                        .sendEvent(item);
                item.setStatus(Event.Status.EXECUTION);
            });
        } else {
            EventSource eventSource = (EventSource) Optional.ofNullable(getEventSourceManager().getEventSource(event.getEventSourceName()))
                    .orElseThrow(() -> new NullPointerException("未获取到事件处理器配置项"));

            Optional.ofNullable(eventSource.getChannelProvider())
                    .orElseThrow(() -> new NullPointerException("事件处理器对应的发送通道为空"))
                    .sendEvent(event);
        }
        event.setStatus(Event.Status.EXECUTION);
        event.persistence();
    }

    /**
     * 获取事物项
     *
     * @param transactionId
     * @return
     */
    private TransactionItem getTransactionItem(@NonNull String transactionId) {
        TransactionItem transactionItem = eventTransactionMap.get(transactionId);
        if (transactionItem == null) {
            throw new IllegalStateException("the specified transaction is not turned on or has been executed.");
        }
        return transactionItem;
    }

    @ToString
    @EqualsAndHashCode
    private class TransactionItem {
        /**
         * 是否已提交成功
         */
        private volatile boolean isCommited = false;
        /**
         * 是否已执行回滚操作
         */
        private volatile boolean isRollback = false;
        /**
         * 事物事件队列
         */
        @Getter
        private final AggregationEvent aggregationEvent = new AggregationEvent.EventBuilder().build();


        public boolean isCommited() {
            return isCommited;
        }

        public synchronized void rollback() {
            aggregationEvent.frozenModify(false);
            isRollback = true;
        }

        public synchronized boolean commit() {
            if (isRollback) {
                throw new IllegalStateException("上下文事件管理器已执行回滚操作，不能再执行持久化操作");
            }
            persistence();
            if (!isCommited) {
                isCommited = true;
                return doCommit();
            } else {
                throw new IllegalStateException("上下文事件管理器已执行事物提交");
            }
        }

        /**
         * 持久化事件操作
         *
         * @return
         */
        protected boolean persistence() {
            aggregationEvent.frozenModify(true);
            return aggregationEvent.persistence();
        }

        /**
         * 执行提交动作
         *
         * @return
         */
        protected boolean doCommit() {
            DefaultEventBus.this.sendEvent(aggregationEvent);
            return true;
        }
    }
}
